package org.com.blbl.Province;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Collectors;

public class ProvinceReducer extends Reducer<Text, IntWritable, Text,  Text> {

    private Map<String, Integer> provinceCountMap = new HashMap<>();
    private int totalCount = 0;

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context){
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        provinceCountMap.put(key.toString(), sum);
        totalCount += sum;
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 按出现次数排序
        provinceCountMap = provinceCountMap.entrySet()
                .stream()
                .sorted((e1, e2) -> e2.getValue().compareTo(e1.getValue()))
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        Map.Entry::getValue,
                        (e1, e2) -> e1,
                        LinkedHashMap::new
                ));

        // 输出格式: "省份名 占比："
        for (Map.Entry<String, Integer> entry : provinceCountMap.entrySet()) {
            double percentage = (entry.getValue() * 100.0) / totalCount;
            context.write(new Text(entry.getKey()), new Text("占比：" + String.format("%.2f%%", percentage)));
        }
    }
}

